package springboot.kafka.controller;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.graylog2.syslog4j.Syslog;
import org.graylog2.syslog4j.SyslogIF;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import springboot.kafka.util.DateUtils;
import springboot.kafka.util.UDPSend;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @author chenming
 * Date  2022/11/14 15:45
 * @version 1.0
 * description
 */

@RestController
@Slf4j
public class TestController {

    @PostMapping("/testKafka1")
    public void test1(@RequestBody JSONObject jsonObject) {
        log.info("数据开始发送");
        int num = jsonObject.getInteger("num");
        String address = jsonObject.getString("address");
        int port = jsonObject.getInteger("port");
        // int count = jsonObject.getInteger("count");
        //String message = jsonObject.getString("message");
        int repeat = jsonObject.getInteger("repeat");
        for (int j = 0; j < repeat; j++) {
//                long time = System.currentTimeMillis();
            testmain(num, address, port, DateUtils.formatDate(new Date()));
//                long t = System.currentTimeMillis() - time;
//                if(t < 1000L){
//                    try {
//                        Thread.sleep(1000L - t);
//                        log.info("等待了" + (1000L - t) + "ms，发送下一条数据");
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }

        }
        log.info("数据发送结束");

    }

    @PostMapping("/testKafka2")
    public void test2(@RequestBody JSONObject jsonObject) {
        int num = jsonObject.getInteger("num");
        String address = jsonObject.getString("address");
        int port = jsonObject.getInteger("port");
        int count = jsonObject.getInteger("count");
        String message = jsonObject.getString("message");
        int repeat = jsonObject.getInteger("repeat");
        for (int j = 0; j < repeat; j++) {
            for (int i = 0; i < count; i++) {
                testmain(num, address, port, message);
            }
        }
    }

    @PostMapping("/testKafka3")
    public void test3(int num, String address, int port, String message) {
        testmain(num, address, port, message);
    }

    @PostMapping("/testKafka4")
    public void test4(int num, String address, int port, String message) {
        testmain(num, address, port, message);
    }

    @PostMapping("/testKafka5")
    public void test5(int num, String address, int port, String message) {
        testmain(num, address, port, message);
    }

//    public void testmain(int num, String address, int port, String message) {
//        new Thread(() -> {
//            List<String> messageList = new ArrayList<>();
//            for (int i = 0; i < num; i++) {
//                messageList.add(message);
//            }
//            UDPSend.generate(messageList, address, port);
//        }).start();
//    }

    public void testmain(int num, String address, int port, String message) {
        for (int i = 0; i < num / 100; i++) {
            new Thread(() -> {
                List<String> messageList = new ArrayList<>();
                for (int j = 0; j < 100; j++) {
                    messageList.add(message);
                }
                UDPSend.generate(messageList, address, port);
            }).start();
        }
    }


    @PostMapping("/testKafka")
    public void test(@RequestBody JSONObject jsonObject) throws Exception {
        Long curTime = System.currentTimeMillis();
        String str = jsonObject.getString("str");
        int time = jsonObject.getInteger("time");
        curTime = curTime + time * 1000;
        final Long[] nowTime = {System.currentTimeMillis()};
        SyslogIF syslog = Syslog.getInstance("udp");
        syslog.getConfig().setHost("192.168.182.226");
        syslog.getConfig().setPort(55140);
        log.info("数据开始发送");
        for (int i = 0; i < 10; i++) {
            Long finalCurTime = curTime;
            new Thread(() -> {
                while (nowTime[0] < finalCurTime) {
                    nowTime[0] = System.currentTimeMillis();
                    try {
                        syslog.log(0, URLDecoder.decode(str, "utf-8"));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }

        log.info("数据发送结束");

    }

}
